package com.mjf.gmall.realtime.util

import java.util

import io.searchbox.client.config.HttpClientConfig
import io.searchbox.client.{JestClient, JestClientFactory}
import io.searchbox.core.{Bulk, BulkResult, Index, Search, SearchResult}
import org.elasticsearch.index.query.{BoolQueryBuilder, MatchQueryBuilder}
import org.elasticsearch.search.builder.SearchSourceBuilder

/**
 * 用来操作ES的工具类
 */
object MyEsUtil {

  // 连接ES的连接池
  var factory : JestClientFactory = null;

  def getClient:JestClient ={
    if(factory==null)build();
    factory.getObject

  }

  def  build(): Unit ={
    factory=new JestClientFactory
    factory.setHttpClientConfig(new HttpClientConfig.Builder("http://hadoop103:9200" )
      .multiThreaded(true)
      .maxTotalConnection(20)
      .connTimeout(10000)
      .readTimeout(10000)
      .build())

  }

  def addDoc(): Unit ={   // 保存数据
    val jest: JestClient = getClient

    // build 设计模式
    // 可转化为json的对象
    val index: Index = new Index.Builder(MovieTest("0104", "lucy"))
        .index("movie_test_20200618")
        .`type`("_doc")
        .id("0104")
        .build()

    val message: String = jest.execute(index).getErrorMessage

    if(message != null) {
      println(message)
    }

    jest.close()
  }

  def bulkDoc(sourceList : List[(String, Any)], indexName : String): Unit = {   // 批量保存数据
    if(sourceList != null && sourceList.size > 0) {
      val jest: JestClient = getClient

      val bulkBuilder: Bulk.Builder = new Bulk.Builder   // 构造批次操作

      for((id, source) <- sourceList) {
        val index: Index = new Index.Builder(source)
          .index(indexName)
          .`type`("_doc")
          .id(id)   // 进行幂等性操作
          .build()
        bulkBuilder.addAction(index)
      }

      val bulk: Bulk = bulkBuilder.build()
      val result: BulkResult = jest.execute(bulk)
      val items: util.List[BulkResult#BulkResultItem] = result.getItems
      println("保存到ES:" + items.size() + "条数")

      jest.close()
    }
  }

  def query(): Unit ={   // 查询数据
    val jest: JestClient = getClient

    val query =
      """
        |{
        |  "query": {
        |    "bool": {
        |      "must": [
        |        {
        |          "match": {
        |            "name": "lucy"
        |          }
        |        }
        |      ]
        |    }
        |  }
        |
        |}
        |""".stripMargin

    val ssb: SearchSourceBuilder = new SearchSourceBuilder()

    val bqb: BoolQueryBuilder = new BoolQueryBuilder()
    bqb.must(new MatchQueryBuilder("name", "lucy"))

    val query2: String = ssb.query(bqb).toString
    println(query2)

    val search: Search = new Search.Builder(query2)
      .addIndex("movie_test-query")
      .addType("_doc")
      .build()

    val result: SearchResult = jest.execute(search)

    val rsList: util.List[SearchResult#Hit[MovieTest, Void]] = result.getHits(classOf[MovieTest])

    import scala.collection.JavaConversions._

    val sourceList: List[MovieTest] = rsList.map(_.source).toList

    sourceList.foreach(println(_))

    jest.close()
  }

  def main(args: Array[String]): Unit = {

    addDoc()

    query()

  }

  case class MovieTest(id: String, name : String)


}
